package com.hngy.java.tablesql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * TableAPI 和 SQL的使用
 */
public class TableAPIAndSQLOpJava {

    public static void main(String[] args) {
        //获取TableEnvironment
        EnvironmentSettings sSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment sTableENV = TableEnvironment.create(sSettings);

        //创建输入表
        sTableENV.executeSql("" +
            "create table myTable(\n" +
            "id int,\n" +
            "name string\n" +
            ") with (\n" +
            "'connector.type' = 'filesystem',\n" +
            "'connector.path' = 'D:\\data\\source',\n" +
            "'format.type' = 'csv'\n" +
            ")");


        //使用Table API实现数据查询和过滤等操作
        /*Table res = sTableENV.from("myTable")
                .select($("id"), $("name"))
                .filter($("id").isGreater(1));

        res.execute().print();*/

        //使用SQL实现数据查询和过滤等操作
        Table res = sTableENV.sqlQuery("select id,name from myTable where id > 1");
        res.execute().print();

        //创建输出表
        sTableENV.executeSql("" +
                "create table newTable(\n" +
                "id int,\n" +
                "name string\n" +
                ") with (\n" +
                "'connector.type' = 'filesystem',\n" +
                "'connector.path' = 'D:\\data\\source1',\n" +
                "'format.type' = 'csv'\n" +
                ")");
        res.executeInsert("newTable");
    }
}
